﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Text;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    // Base class representing an IActorProxy object.  FabricActorProxy and other platform-specific IActorProxy classes derive from this class.
    public class ActorProxyBase : IActorProxy, IDisposable
    {
        private readonly ActorClient m_client;
        private readonly IActorState m_state;
        private readonly string m_targetPartitionName; // friendly partition name for target
        private readonly string m_myPartitionName; // client's partition name

        // We need this registry here so that IActorState implementations can register themselves with FabricActorProxy,
        // so that our magic IDisposable implementation can obtain actor state given an actor name.
        internal static Dictionary<string, IActorState> s_stateMapper = new Dictionary<string, IActorState>();
        internal static void RegisterState(string serviceName, string partitionName, IActorState state)
        {
            var key = serviceName + ":" + partitionName;
            lock (s_stateMapper) s_stateMapper[key] = state;
        }
        public static void UnregisterState(string serviceName, string partitionName)
        {
            var key = serviceName + ":" + partitionName;
            lock (s_stateMapper) s_stateMapper.Remove(key);
        }

        public ActorProxyBase(ActorClient client, IActorState sourceState, string targetPartitionName = null, string myPartitionName = null)
        {
            Contract.Requires(client != null);
            Contract.Requires(sourceState != null);

            m_client = client;
            m_state = sourceState;
            m_targetPartitionName = targetPartitionName;
            m_myPartitionName = myPartitionName;
        }

        public void Command(string methodName, object[] parameters)
        {
            m_client.CallMethodFireAndForget(methodName, parameters);
        }

        public IActorPromise Request(string methodName, object[] parameters)
        {
            var completionTask = m_client.CallMethodAsyncRaw(methodName, parameters);
            return new ActorPromise(m_state, completionTask);
        }

        private readonly Dictionary<string, IActorStateObservable> m_observableCache = new Dictionary<string, IActorStateObservable>();
        public IActorStateObservable this[string eventType]
        {
            get
            {
                IActorStateObservable result;
                if (!m_observableCache.TryGetValue(eventType, out result))
                {
                    result = new ActorStateObservable(eventType, m_client, m_state, m_targetPartitionName, m_myPartitionName);
                    m_observableCache[eventType] = result;
                }
                return result;
            }
        }

        // Shut down the actor client
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                // TODO: There may me more things to clean up here.  But the fabric actor client is a good start.

                // Clean up the cache of observables.
                foreach (var eventObservable in m_observableCache)
                    eventObservable.Value.Unsubscribe(eventObservable.Key);
                m_observableCache.Clear();

                // It's a little odd to dispose of something that was passed into your constructor,
                // but in this case m_client was constructed on its way in, and so we can properly
                // dispose of it here.
                m_client.Dispose();
            }
        }

        public bool AddAssembly(String assemblyName, byte[] assemblyBytes)
        {
            return m_client.AddAssembly(assemblyName, assemblyBytes);
        }

        // This is what is returned by ClientWrapper[eventType].
        // It allows you to subscribe & unsubscribe to events of a certain type from a certain client.
        class ActorStateObservable : IActorStateObservable
        {
            private readonly string m_eventType;
            private readonly ActorClient m_client;
            private readonly IActorState m_state;
            private readonly string m_actorName;
            private readonly string m_targetPartitionName; // target actor partition name
            private readonly string m_myPartitionName; // client partition key
            private ActorStateObserver m_observer;
            private int m_subscriptionCount = 0;

            public ActorStateObservable(string eventType, ActorClient client, IActorState state, string targetPartitionName, string myPartitionName)
            {
                Contract.Requires(eventType != null);
                Contract.Requires(client != null);
                Contract.Requires(state != null);

                m_eventType = eventType;
                m_client = client;
                m_state = state;
                m_actorName = client.ActorUri.ToString();
                m_targetPartitionName = targetPartitionName;
                m_myPartitionName = myPartitionName;
            }

            // It is assumed that Subscribe/Unsubscribe are called from either actor methods
            // or event handlers (which are just actor methods, but called in response to an event
            // reception rather than an explicit user request).  In both of these situations, the
            // method is called under state-lock, which should mean that we don't need any additional
            // locking.

            // Allow for the holder of an observable to specify a handler for it.  Recall that
            // the external actor name and the event type were already specified in creating the
            // observable.
            public IDisposable Subscribe(string handlerName)
            {
                // Record this subscription in the state, so that it can be rehydrated if the primary fails over
                var key = Utilities.UtilityMethods.CreateEventHandlerKey(m_actorName, m_eventType, m_targetPartitionName);
                m_state.Set(key, handlerName);

                // Fire up your observer and subscribe to the client, if you have not yet done so
                if (m_observer == null)
                {
                    m_observer = new ActorStateObserver(m_state, m_actorName, m_targetPartitionName);
                    m_client.Subscribe(m_eventType, m_observer);
                }

                ActorETWEventSource.Log.SubscriptionRequestSent(m_state.GetId(), m_myPartitionName, m_actorName, m_targetPartitionName, m_eventType);
                m_subscriptionCount++;
                return new SubscriptionDisposable(m_state.GetId(), m_myPartitionName, m_actorName, m_targetPartitionName, m_eventType, handlerName);
            }

            // Allow for explicit unsubscription
            public void Unsubscribe(string handlerName)
            {
                var key = Utilities.UtilityMethods.CreateEventHandlerKey(m_actorName, m_eventType, m_targetPartitionName);
                m_state.Remove(key);

                // Or we could just leave it subscribed so that we don't have to do a client subscribe on
                // the next subscribe.
                m_subscriptionCount--;
                if (m_subscriptionCount == 0)
                {
                    m_client.UnsubscribeFireAndForget(m_eventType);
                    m_observer = null;
                }
            }

        }

        // This is the observer registered with the client.
        class ActorStateObserver : IObserver<ActorEvent>
        {
            private IActorState m_state;
            private string m_actorName;
            private string m_targetPartitionName;

            public ActorStateObserver(IActorState actorState, string actorName, string targetPartitionName)
            {
                Contract.Requires(actorState != null);
                Contract.Requires(!String.IsNullOrWhiteSpace(actorName));

                m_state = actorState;
                m_actorName = actorName;
                m_targetPartitionName = targetPartitionName;
            }

            public void OnCompleted()
            {
            }

            public void OnError(Exception error)
            {
                Trace.WriteLine("Error detected in ActorStateObserver.  {0}", error.ToString());
                //throw new NotImplementedException();
            }

            public void OnNext(ActorEvent value)
            {
                var eventType = value.EventType;
                var key = Utilities.UtilityMethods.CreateEventHandlerKey(m_actorName, eventType, m_targetPartitionName);
                value.EventSource = m_actorName;
                value.EventSourcePartitionName = m_targetPartitionName;
                ActorETWEventSource.Log.EventReceived(m_state.GetId(), m_state.PartitionName, value);

                // TODO: We could just call the method & let it run.  Joe believes this would be efficient, but it
                // could lead to out of order events.  That probably isn't safe, if we have events that depend on order
                // ie, processing events like "add, add, remove" vs. "add, remove, add" can be very different.
                // Does Rx provide any ordering guarantee for us, or if it does, do we break it by doing an async method call
                // and not blocking for the result here?
                Task t = MethodCallUtilities.CallMethodOnStateAsync(m_state, key, true, value);
                t.Wait();
            }
        }

        // Supports the Rx convention of unsubscription-via-disposal.  This Disposable can be safely
        // cached and re-hydrates correctly on new primaries.
        [Serializable]
        class SubscriptionDisposable : IDisposable
        {
            private string m_myName;
            private string m_myPartitionName;
            private string m_otherActorName;
            private string m_targetPartitionName;
            private string m_eventType;
            private string m_handlerName;

            public SubscriptionDisposable(string myName, string myPartitionName, string otherActorName, string targetPartitionName, string eventType, string handlerName)
            {
                Contract.Requires(!String.IsNullOrWhiteSpace(myName));
                Contract.Requires(!String.IsNullOrWhiteSpace(otherActorName));
                Contract.Requires(!String.IsNullOrWhiteSpace(handlerName));

                m_myName = myName;
                m_myPartitionName = myPartitionName;
                m_otherActorName = otherActorName;
                m_eventType = eventType;
                m_handlerName = handlerName;
                m_targetPartitionName = targetPartitionName;
            }

            public void Dispose()
            {
                var state = s_stateMapper[m_myName+":"+m_myPartitionName];
                var client = state.GetActorProxy(m_otherActorName, m_targetPartitionName);
                var observable = client[m_eventType];
                observable.Unsubscribe(m_handlerName);
            }
        }

    }
}
